package com.itheima.stock.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.itheima.stock.entity.MessageEntity;
import com.itheima.stock.entity.OrderEntity;
import com.itheima.stock.entity.StockEntity;
import com.itheima.stock.mapper.MessageMapper;
import com.itheima.stock.service.StockService;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(topic = "order-topic",
        consumerGroup = "stock-service-group",
        selectorExpression = "stock",
        consumeTimeout = 1L)
//selectorExpression = "" , 过滤tag标签,只接收tag="stock"的消息
//consumeMode = ConsumeMode.ORDERLY ,指定消费者要有序获取消息
//consumeTimeout = 1L , 消费者超时1分钟
public class OrderMessageListener implements RocketMQListener<Message> {

    @Autowired
    private StockService stockService;

    @Autowired
    private MessageMapper messageMapper;

    @Override
    public void onMessage(Message message) {

        System.out.println(LocalTime.now() + "stock service : " + new String(message.getBody()));

        //获取并解析消息内容
        //keys: 唯一标识
        String keys = message.getKeys();

        MessageEntity messageEntity = new MessageEntity();
        messageEntity.setMessageId(keys);

        try {
            messageMapper.insert(messageEntity);
        } catch (DuplicateKeyException e) {
            System.out.println("该消息正在被处理,不做任何处理返回");
            return;
        }

        //模拟超时情况
        try {
            //超时2分钟
            TimeUnit.MINUTES.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //解析消息
        OrderEntity orderEntity = JSON.parseObject(new String(message.getBody()), OrderEntity.class);

        QueryWrapper<StockEntity> wrapper = new QueryWrapper<>();
        wrapper.lambda().eq(StockEntity::getGoodsId, orderEntity.getGoodsId());
        StockEntity stockEntity = stockService.getOne(wrapper);

        stockService.reduceStock(orderEntity.getGoodsId(), orderEntity.getNumber(), stockEntity.getVersion());

        //删除消息防重表的messageId
        messageMapper.deleteById(keys);

    }
}
